package com.hashimoto.core.remote.client.edge;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

import com.hashimoto.core.handler.business.EdgeHandler;
import com.hashimoto.core.handler.decodec.CodecAdapter;
import com.hashimoto.core.message.CallBackSignal;
import com.hashimoto.core.message.EdgeInitiativePushSignal;
import com.hashimoto.core.message.RegistryEdgeSignal;
import com.hashimoto.core.remote.client.ClientConfig;
import com.hashimoto.core.remote.client.ClientType;
import com.hashimoto.core.remote.exchange.CallMessage;
import com.hashimoto.core.util.GsonUtil;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;

public class NettyEdgeClient extends EdgeClient {

	private NioEventLoopGroup nioEventLoopGroup;

	private Bootstrap bootstrap;

	private volatile Channel channel; // volatile, please copy reference to use

	private ClientConfig config;

	private EdgeProcessHandler hander;

	private EdgeCallback callBack;
	
	@Override
	public void clientConfig(ClientConfig config) {
		this.config = config;
	}

	@Override
	public SocketAddress remoteAddress() {
		return channel != null ? channel.remoteAddress() : null;
	}

	@Override
	public void excepton(Throwable t) {

	}

	@SuppressWarnings("rawtypes")
	@Override
	public void start() {

		hander = new NettyEdgeProcessHandler();
		hander.edgeCallback(callBack);
		hander.setConfig(config);
		bootstrap = new Bootstrap();
		nioEventLoopGroup = new NioEventLoopGroup(this.config.getThreadNum(),
				new DefaultThreadFactory("nettyEdgeWorker", true));

		bootstrap.group(nioEventLoopGroup).option(ChannelOption.SO_KEEPALIVE, true)
				.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
				.channel(NioSocketChannel.class);

		if (this.config.getConnectTimeout() < 3000) {
			bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
		} else {
			bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.config.getConnectTimeout());
		}

		bootstrap.handler(new ChannelInitializer() {
			@Override
			protected void initChannel(Channel ch) throws Exception {
				CodecAdapter adapter = new CodecAdapter();
				ch.pipeline()
						.addLast("ping", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS))
						.addLast("decoder", adapter.getDecode()).addLast("encoder", adapter.getEncode())
						.addLast("handler", new EdgeHandler(hander));
			}
		});

		bootstrap.remoteAddress(this.config.getRemoteHost(), this.config.getRemotePort());
		ChannelFuture future = bootstrap.connect();
		boolean ret = future.awaitUninterruptibly(this.config.getConnectTimeout(), TimeUnit.MILLISECONDS);
		if (ret && future.isSuccess()) {
			this.channel = future.channel();
		}

	}

	@Override
	public void stop() {
		if (nioEventLoopGroup != null) {
			nioEventLoopGroup.shutdownGracefully();
		}
	}

	@Override
	public boolean isActive() {
		return false;
	}

	@Override
	public SocketAddress localAddress() {
		return channel != null ? channel.localAddress() : null;
	}

	@Override
	public void send(String interfaceName, Long msgId, int code, int retweetsId, Object data) {
		CallBackSignal signal = new CallBackSignal();
		signal.setInterfaceName(interfaceName);
		signal.setParkId(this.config.getClientId());
		signal.setMsgId(msgId);
		signal.setCode(code);
		signal.setCallMode((byte) 0);
		signal.setRetweetsId(retweetsId);
		signal.setData(data != null ? GsonUtil.toJson(data).getBytes() : new byte[0]);
		channel.writeAndFlush(signal);

	}

	@Override
	public void report(int eventKey, Object data) {
		EdgeInitiativePushSignal signal = new EdgeInitiativePushSignal();
		signal.setParkId(this.config.getClientId());
		signal.setEventKey(eventKey);
		signal.setMsgId(System.currentTimeMillis());
		signal.setData(data != null ? GsonUtil.toJson(data).getBytes() : new byte[0]);
		channel.writeAndFlush(signal);
	}

	@Override
	public void setReceivedCallback(EdgeCallback callback) {
		this.callBack = callback;
	}

	@Override
	public void registry(Object data) {
		RegistryEdgeSignal signal = new RegistryEdgeSignal();
		signal.setData(GsonUtil.toJson(data).getBytes());
		signal.setEdgeId(this.config.getClientId());
		channel.writeAndFlush(signal);	
	}

}
